package org.example.data_stream;

import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.ArrayList;
import java.util.List;

/**
 * 聚合
 * @author shenguangyang
 */
public class E07_TransformSimpleAgg {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        List<Event> events = new ArrayList<Event>() {{
            add(new Event("user01", "./home", 1000L));
            add(new Event("user02", "./cart", 2000L));
            add(new Event("user02", "./prod?id=10`", 3800L));
            add(new Event("user02", "./prod?id=2`", 3000L));
            add(new Event("user01", "./prod?id=15`", 4000L));
        }};
        DataStream<Event> streamSource = env.fromCollection(events);

        // 按键分组之后进行聚合, 提取当前用户最近一次访问数据
        // max只对当前字段取最大值,其他字段直接采用当前记录数据
        // maxBy取当前字段最大的那条记录
        streamSource.keyBy(new KeySelector<Event, String>() {
            @Override
            public String getKey(Event event) throws Exception {
                return event.getUser();
            }
        }).max("timestamp").print("max: ");

        streamSource.keyBy(Event::getUser).maxBy("timestamp").print("maxBy: ");

        env.execute();
    }
}
